﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Linq;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.Samples.ServiceHosting.StorageClient;
using System.Data.Services.Client;
using System.Net;
using System.Xml;
using System.Xml.Linq;
using TweetPollDataService;
using System.Diagnostics;

namespace TwitterPoll_WorkerRole
{
    public class WorkerRole : RoleEntryPoint
    {
        const string PartitionKey = "tweetLengthCount";

        public override bool OnStart()
        {
            // This is a sample worker implementation. Replace with your logic.
            //RoleManager.WriteToLog("Information", "Worker Process entry point called");
            Trace.WriteLine("Worker Process entry point called","Information");
            

            Thread getTweets = new Thread(new ThreadStart(GetTweetsThreadController));
            getTweets.Start();

            Thread storeTweets = new Thread(new ThreadStart(StoreTweetsThreadController));
            storeTweets.Start();
            return base.OnStart();
        }

        private static void StoreTweetsThreadController()
        {
            while (true)
            {
                try
                {
                    StoreTweetInTable("tweetpoll");
                }
                catch (Exception e)
                {
                    //RoleManager.WriteToLog("Error", "General error: " + e.Message);
                    Trace.WriteLine("General error: " + e.Message, "Error");
                }
                finally
                {
                    Thread.Sleep(1500);
                }
            }
        }

        private static void StoreTweetInTable(string queueName)
        {
            bool exists;

            //Step 1: Get a message off the queue
            StorageAccountInfo queueAccount = StorageAccountInfo.GetDefaultQueueStorageAccountFromConfiguration();
            QueueStorage queueService = QueueStorage.Create(queueAccount);
            queueService.RetryPolicy = RetryPolicies.RetryN(2, TimeSpan.FromMilliseconds(100));
            MessageQueue q = queueService.GetQueue(queueName);
            q.CreateQueue(out exists);

            Message msg = q.GetMessage();

            //Step 2: Increment tweet length counter
            if (msg != null)
            {
                StorageAccountInfo tableAccount = StorageAccountInfo.GetDefaultTableStorageAccountFromConfiguration();
                TweetPollDataServiceContext svc = new TweetPollDataServiceContext(tableAccount);


                XElement el = XElement.Parse(msg.ContentAsString(), LoadOptions.PreserveWhitespace); 
                int tweetLength = el.Element("text").Value.Length;
                
                //Does a counter already exist for this tweetlength
                var resultQuery = from row in svc.Table
                                  where row.RowKey == tweetLength.ToString() && row.PartitionKey == PartitionKey
                                  select row;
                
                TableStorageDataServiceQuery<TweetPollEntity> query = new TableStorageDataServiceQuery<TweetPollEntity>(resultQuery as DataServiceQuery<TweetPollEntity>, svc.RetryPolicy);
                IEnumerable<TweetPollEntity> results = query.ExecuteAllWithRetries();
                if (results.Count() == 0)
                {
                    //add a new entity (i.e. row)
                    svc.AddObject(TweetPollDataServiceContext.TableName, new TweetPollEntity(PartitionKey, tweetLength.ToString(), 1));
                }
                else
                {
                    //or increment existing
                    var entity = results.SingleOrDefault();
                    entity.Count = ++entity.Count;
                    svc.UpdateObject(entity);
                }
                svc.SaveChangesWithRetries();
                
                //Step 3: Record the message if it is longer than 140
                if (tweetLength > 140)
                {
                    StoreLongTweet(el.Element("text").Value, svc);
                }

                //Step 4: We've dealed with the message so delete it from the queue!
                q.DeleteMessage(msg);
            }
        }

        private static void StoreLongTweet(string tweet, TweetPollDataServiceContext svc)
        {
            svc.AddObject(TweetPollDataServiceContext.LongTweetsTableName, new LongTweetEntity(tweet));
            svc.SaveChangesWithRetries();
        }


        private static void GetTweetsThreadController()
        {
            while (true)
            {
                try
                {
                    // Get the public timeline, caching the result for two minutes
                    HttpWebRequest request = (HttpWebRequest)WebRequest.Create("http://twitter.com/statuses/public_timeline.xml ");
                    HttpWebResponse response2 = (HttpWebResponse)request.GetResponse();
                    XmlReader xmlReader = XmlReader.Create(response2.GetResponseStream());
                    XDocument xmlDoc = XDocument.Load(xmlReader);
                    var tweets = from el in xmlDoc.Descendants("status")
                                 select el;

                    //RoleManager.WriteToLog("Information", tweets.Count() + " tweets returned at " + DateTime.Now);

                    PutTweetsOnQueue(tweets, "tweetpoll");
                }
                catch (WebException we)
                {
                    //RoleManager.WriteToLog("Error", "Network error: " + we.Message);
                    Trace.WriteLine("Network error: " + we.Message, "Error");
                }
                catch (StorageException se)
                {
                    //RoleManager.WriteToLog("Error", "Storage service error: " + se.Message);
                    Trace.WriteLine("Storage service error: " + se.Message, "Error");
                }
                catch (Exception e)
                {
                    //RoleManager.WriteToLog("Error", "General error: " + e.Message);
                    Trace.WriteLine("General error: " + e.Message, "Error");
                }
                finally
                {
                    Thread.Sleep(60000);
                }
            }
        }

        private static void PutTweetsOnQueue(IEnumerable<XElement> tweets, string queueName)
        {
            //Valid queue name documentation at http://msdn.microsoft.com/en-us/library/dd179349.aspx
            bool exists;

            //Get a ref to the queue and create it if it doesn't already exist
            StorageAccountInfo account = StorageAccountInfo.GetDefaultQueueStorageAccountFromConfiguration();
            QueueStorage queueService = QueueStorage.Create(account);
            queueService.RetryPolicy = RetryPolicies.RetryN(2, TimeSpan.FromMilliseconds(100));
            MessageQueue q = queueService.GetQueue(queueName);
            q.CreateQueue(out exists);
            //q.Clear();

            //put each tweet onto the queue one at a time
            foreach (var tweet in tweets)
            {
                Message msg = new Message(String.Format("{0}", tweet));
                q.PutMessage(msg);
            }
            //var allMsgsInQ = from t in q.PeekMessages(32) select t.ContentAsString();
        }

        //public override RoleStatus GetHealthStatus()
        //{
        //    // This is a sample worker implementation. Replace with your logic.
        //    return RoleStatus.Healthy;
        //}
    }

    
}
